package com.poetic.market.analysis.source;

import com.poetic.market.analysis.domain.MarketingUserBehavior;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class SimulatedEventSource extends RichParallelSourceFunction<MarketingUserBehavior> {

    private Boolean running = Boolean.TRUE;

    List<String> channelList = Arrays.asList("AppStore", "XiaomiStore", "HuaweiStore", "weibo",
            "wechat", "tieba");

    List<String> behaviorTypeList =  Arrays.asList("BROWSE", "CLICK", "PURCHASE", "UNINSTALL");


    private Random random = new Random();

    @Override
    public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
        Long maxExements = Long.MAX_VALUE;

        Long count = 0L;

        while (running && count < maxExements) {


            String id = UUID.randomUUID().toString();
            String behaviorType = behaviorTypeList.get(random.nextInt(behaviorTypeList.size()));

            String channel = channelList.get(random.nextInt(channelList.size()));
            long ts = System.currentTimeMillis();
            MarketingUserBehavior behavior = MarketingUserBehavior.builder()
                    .userId(id)
                    .behavior(behaviorType)
                    .channel(channel)
                    .timestamp(ts)
                    .build();
            ctx.collectWithTimestamp(behavior, ts);
            count += 1;
            TimeUnit.MICROSECONDS.sleep(5L);

        }
    }

    @Override
    public void cancel() {
        running = Boolean.FALSE;
    }
}
